package com.atguigu.flink0922.chapter11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.hive.HiveCatalog;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/3/12 11:09
 */
public class Flink11_Catalog_Hive {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        final String my_hive = "my_hive";
        final String db = "flink_0922";
        // 1. 创建catalog对象
        final HiveCatalog hiveCatalog = new HiveCatalog(my_hive, db, "input");
        // 2. 在环境中注册catalog
        tEnv.registerCatalog(my_hive, hiveCatalog);
        
        // 3. 把hive catalog设置默认
        tEnv.useCatalog(my_hive);
        tEnv.useDatabase(db);
        
        tEnv.sqlQuery("select * from person").execute().print();
        
    }
}
